package com.macaline.technology.rabbitMq;

import com.rabbitmq.client.*;
import com.rabbitmq.client.AMQP.BasicProperties;

import java.io.IOException;
import java.util.concurrent.TimeoutException;


public class Test {

	public static String ip = "127.0.0.1"; // IP地址
	public static int port = 5672; // 端口号
	public static String username = "admin"; // 用户名
	public static String password = "admin"; // 密码
	public static String virtualHost = "/admin";

	private static Connection connection = null;


	/**
	 * 获取MQ连接工具
	 *
	 * @author warm_joy
	 * @return
	 * @throws IOException
	 * @throws TimeoutException
	 */
	public static Connection getConnection() throws IOException, TimeoutException {
		if (connection == null || connection.isOpen() == false) {
			ConnectionFactory factory = new ConnectionFactory();
			factory.setHost(ip);
			factory.setPort(port);
			factory.setVirtualHost(virtualHost);
			factory.setUsername(username);
			factory.setPassword(password);
			factory.setAutomaticRecoveryEnabled(true);
			connection = factory.newConnection();
		}
		return connection;
	}

	public static void main(String[] args) throws IOException, TimeoutException {

		Connection connection = getConnection();
		Channel channel = connection.createChannel();
		channel.confirmSelect();
		channel.queueDeclare("test_queue", true, false, false, null);

		String msg = "12345";
		System.out.println("发送消息：" + msg);
		channel.basicPublish("", "test_queue", true, null, msg.getBytes());
		channel.addConfirmListener(new ConfirmListener() {
			@Override
			public void handleNack(long deliveryTag, boolean multiple) throws IOException {
				System.out.println("消息发送");
			}

			@Override
			public void handleAck(long deliveryTag, boolean multiple) throws IOException {
				System.out.println("消息发送成功");
			}
		});

		//添加一个 return 监听
		channel.addReturnListener(new ReturnListener() {
			@Override
			public void handleReturn(int replyCode, String replyText, String exchange, String routingKey,
					AMQP.BasicProperties properties, byte[] body) throws IOException {
				System.out.println("return relyCode: " + replyCode);
				System.out.println("return replyText: " + replyText);
				System.out.println("return exchange: " + exchange);
				System.out.println("return routingKey: " + routingKey);
				System.out.println("return properties: " + properties);
				System.out.println("return body: " + new String(body));
			}
		});

//		channel.queueDeclare("test_queue", true, false, false, null);
		DefaultConsumer consumer = new DefaultConsumer(channel) {
			@Override
			public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
					throws IOException {
				String msg = new String(body);
				System.out.println("接受参数：" + msg);
				// 消费失败通知
				channel.basicNack(envelope.getDeliveryTag(), true, true);


				// 消费成功通知
//				channel.basicAck(envelope.getDeliveryTag(), true);
			}
		};
		channel.basicConsume("test_queue", false, consumer);

		// 关闭资源
//		channel.close();
//		connection.close();
	}

}
